import ast
import json

import numpy as np
import pandas as pd
import pathway as pw

from application.db.mysql_db.nsfc.NsfcInfoList import NsfcInfoList
from application.db.mysql_db.nsfc.NsfcInfoSectionList import NsfcInfoSectionList
from application.db.mysql_db.nsfc.NsfcInfoTypeDict import NsfcInfoTypeDict
from application.db.mysql_db.nsfc.NsfcPublishProjectCodeDict import NsfcPublishProjectCodeDict
from application.db.mysql_db.nsfc.NsfcResourceSourceDict import NsfcResourceSourceDict
from application.etl.base_etl import BaseEtl
from application.etl.write.nsfc_to_es_write import NsfcToEsWrite
from application.etl.write.print_write import PrintWriter
from application.pipelines.es_section_process_pipeline import EsSectionProcessPipeline
from application.settings import ELASTIC_CONNECTION


class NsfcToEsEtlPandas(BaseEtl):
    """
    基于 Pandas 的国自然基金资讯数据 ETL 处理类。

    该类实现从 MySQL 读取资讯主表及相关字典表，利用 Pandas 进行数据转换与清洗，
    并输出统一结构化数据，用于 Elasticsearch 入库或其他下游处理。

    Attributes
    ----------
    pipelines : list
        ETL 处理管道列表，当前仅包含段落处理管道。
    area_filter_default : dict
        默认地区信息。
    apply_code_default : dict
        学部/领域默认值，用于关联失败时填充。
    """
    index_name = "test_information_index"
    pipelines = [EsSectionProcessPipeline()]

    def read(self):
        """
        读取并转换国自然基金资讯数据为结构化表。

        读取流程
        ----------
        1. 从 MySQL 获取主表及字典表数据。
        2. 生成字典表的行级 JSON，以支持高效映射。
        3. 按信息 ID 对段落表分组，形成 `sections` 列。
        4. 将段落信息及各类字典映射到主表。
        5. 填充缺失值，标准化时间字段。
        6. 输出 Pathway 表对象。

        Returns
        -------
        pw.Table
            处理完成的结构化表对象，用于后续 transform 或 write。
        """

        # ---------------------------------------------------------------------
        # 从 MySQL 读取主表及各字典表
        # ---------------------------------------------------------------------
        # ["section_attr", "title_level", "marc_code", "src_text", "dst_text", "media_info"]

        nsfc_main = pd.DataFrame(NsfcInfoList.select().dicts(),
                                 columns=NsfcInfoList._meta.fields.keys())
        nsfc_type_mapping = pd.DataFrame(NsfcInfoTypeDict.select().dicts())
        nsfc_paragraphs = pd.DataFrame(NsfcInfoSectionList.select(
            NsfcInfoSectionList.section_id,
            NsfcInfoSectionList.information_id,
            NsfcInfoSectionList.section_attr,
            NsfcInfoSectionList.title_level,
            NsfcInfoSectionList.marc_code,
            NsfcInfoSectionList.src_text,
            NsfcInfoSectionList.dst_text,
            NsfcInfoSectionList.media_info,
        ).order_by(
            NsfcInfoSectionList.information_id.asc(),
            NsfcInfoSectionList.section_order.asc()
        ).dicts(),columns=["section_id", "information_id", "section_attr", "title_level", "marc_code", "src_text", "dst_text", "media_info"])
        nsfc_field_mapping = pd.DataFrame(NsfcPublishProjectCodeDict.select().dicts())
        nsfc_source_mapping = pd.DataFrame(NsfcResourceSourceDict.select().dicts())

        # ---------------------------------------------------------------------
        # 字典表生成行级 JSON 格式，用于高效映射
        # ---------------------------------------------------------------------
        nsfc_type_mapping["row_json"] = nsfc_type_mapping[["info_type_id", "info_type_name"]].to_dict(orient="records")

        nsfc_source_mapping["source_name_zh"] = nsfc_source_mapping["source_name"].str.get("zh")

        nsfc_source_mapping["row_json"] = nsfc_source_mapping[
            ["source_id", "source_name_zh", "source_main_link"]
        ].rename(columns={"source_name_zh": "source_name"}).to_dict(orient="records")

        nsfc_field_mapping["row_json"] = nsfc_field_mapping[["apply_code", "code_name"]].to_dict(orient="records")

        # 将 section 按 information_id 聚合为列表
        info_section_grouped = nsfc_paragraphs.groupby("information_id").apply(
            lambda x: x.drop(columns=["information_id", "section_id"]).to_dict(orient="records")
            # 去掉 information_id 再转字典列表
        ).to_dict()

        # 将聚合后的列表映射回 info_list
        nsfc_main["sections"] = nsfc_main["information_id"].map(info_section_grouped)

        # 对没有分段信息的填充空列表
        nsfc_main["sections"] = nsfc_main["sections"].apply(lambda x: x if isinstance(x, list) else [])

        # ---------------------------------------------------------------------
        # 使用 map 将字典表信息映射到主表
        # ---------------------------------------------------------------------
        type_map = dict(zip(nsfc_type_mapping["info_type_id"], nsfc_type_mapping["row_json"]))
        source_map = dict(zip(nsfc_source_mapping["source_id"], nsfc_source_mapping["row_json"]))
        field_map = dict(zip(nsfc_field_mapping["apply_code"], nsfc_field_mapping["row_json"]))

        nsfc_main["info_type"] = nsfc_main["info_type_id"].map(type_map).apply(
            lambda x: x if isinstance(x, dict) else {})
        nsfc_main["source_info"] = nsfc_main["source_id"].map(source_map).apply(
            lambda x: x if isinstance(x, dict) else {})
        nsfc_main["apply_info"] = nsfc_main["apply_code"].map(field_map).apply(
            lambda x: x if isinstance(x, dict) else {})

        # ---------------------------------------------------------------------
        # 标准化时间字段，填充缺失值
        # ---------------------------------------------------------------------
        default_date = pd.Timestamp("1970-01-01")
        nsfc_main["publish_date"] = pd.to_datetime(nsfc_main["publish_date"], errors="coerce")
        nsfc_main["publish_date"] = nsfc_main["publish_date"].fillna(default_date)
        nsfc_main["publish_date"] = nsfc_main["publish_date"].dt.strftime("%Y-%m-%d")

        # ---------------------------------------------------------------------
        # 选择最终输出字段
        # ---------------------------------------------------------------------
        nsfc_final = nsfc_main[
            [
                "information_id",
                "info_type",
                "source_info",
                "apply_info",
                "info_name",
                "original_link",
                "publish_date",
                "sections",
            ]
        ]
        # 转换为 Pathway 表对象供下游使用
        input_data = pw.debug.table_from_pandas(nsfc_final)
        return input_data

    def write(self, transform_data):
        """
        将 transform 后的 Pathway 表数据写出。

        当前实现
        ----------
        将数据写入 Elasticsearch，示例中使用本地测试地址。

        可扩展
        ----------
        - 输出至控制台
        - 写入 Kafka
        - 写入其他数据库

        Parameters
        ----------
        transform_data : pw.Table
            已完成 transform 的 Pathway 表对象。
        """
        # ---------------------------------------------------------------------
        # 写入 Elasticsearch
        # ---------------------------------------------------------------------
        pw.io.python.write(transform_data, NsfcToEsWrite())
        # pw.io.elasticsearch.write(
        #     table=transform_data,
        #     host=ELASTIC_CONNECTION['default']['host'],
        #     auth=pw.io.elasticsearch.ElasticSearchAuth.basic(ELASTIC_CONNECTION['default']['user'],
        #                                                      ELASTIC_CONNECTION['default']['password']),
        #     index_name=self.index_name,
        # )
